package sg.edu.nus.iss.stockex;

import java.io.IOException;
import java.util.Iterator;
import java.util.LinkedList;

import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Reducer;

public class StochasticOscillatorReducerWithoutRunningWindow extends Reducer<CodeDateKey, TimestampPriceDataPoint, Text, Text> {

	@Override
	public void reduce(CodeDateKey key, Iterable<TimestampPriceDataPoint> values,
			Context context) throws IOException, InterruptedException {

		Text outKey = new Text();
		Text outVal = new Text();

		double highestHigh = 0;
		double lowestLow = Double.MAX_VALUE;

		TimestampPriceDataPoint lastDataPoint = null;
		for(TimestampPriceDataPoint dataPoint: values)
		{
			if(dataPoint.getHighPrice()>highestHigh)
			{
				highestHigh = dataPoint.getHighPrice();
			}

			if(dataPoint.getLowPrice() < lowestLow)
			{
				lowestLow = dataPoint.getLowPrice();
			}
			lastDataPoint = dataPoint;
		}

		double currentClosePrice = lastDataPoint.getClosePrice();
		double k = 0;
		if (highestHigh - lowestLow == 0) k = 0;
		else k = (currentClosePrice-lowestLow)/(highestHigh - lowestLow)*100;
		
		String strBackDate = lastDataPoint.getDate();
		/*
		outKey.set("Code: " + key.getCode() + ", Date: "
				+ strBackDate);
		String data = "lowestLow: " + lowestLow + "; highestHigh: " + highestHigh + "; k: " + k;

		outVal.set(data);
*/
		
		outKey.set(key.getCode() + "\t" + strBackDate);
		String data = lowestLow + "\t" + highestHigh + "\t" + k;

		outVal.set(data);
		context.write(outKey, outVal);


	} // end reduce

}
